//! # 消息总线长时间运行测试
//!
//! 测试消息总线在长时间运行情况下的稳定性、内存泄漏和性能衰减问题

use rustcloud_bus::prelude::*;
use rustcloud_bus::{
    bus::MessageBus,
    memory::{InMemoryMessageBus},
    message::{Message, MessagePriority},
    config::MessageBusConfig,
};
use serde_json::json;
use std::sync::{Arc, atomic::{AtomicUsize, AtomicBool, Ordering}};
use std::time::{Duration, Instant};
use tokio::time::{sleep, interval};

/// 长时间运行测试辅助工具
struct LongRunningTestHelper;

impl LongRunningTestHelper {
    /// 创建长时间运行配置的消息总线
    async fn create_long_running_bus() -> Arc<InMemoryMessageBus> {
        let config = MessageBusConfig {
            bus_type: rustcloud_bus::config::MessageBusType::Memory,
            max_connections: 5000,
            buffer_size: 100000,
            max_message_size: 1024 * 1024, // 1MB
            enable_metrics: true,
            ..Default::default()
        };
        
        Arc::new(InMemoryMessageBus::with_config(config).await.unwrap())
    }

    /// 启动持续发布任务
    async fn start_continuous_publisher(
        bus: Arc<InMemoryMessageBus>,
        topic_prefix: &str,
        messages_per_second: usize,
        stop_signal: Arc<AtomicBool>
    ) -> Arc<AtomicUsize> {
        let published_count = Arc::new(AtomicUsize::new(0));
        let published_count_clone = published_count.clone();
        let topic_prefix = topic_prefix.to_string();
        
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_millis(1000 / messages_per_second as u64));
            let mut message_id = 0;
            
            while !stop_signal.load(Ordering::Relaxed) {
                interval.tick().await;
                
                let topic = format!("{}.{}", topic_prefix, message_id % 10);
                let message = Message::new(&topic, json!({
                    "id": message_id,
                    "timestamp": chrono::Utc::now().timestamp(),
                    "data": format!("Message {}", message_id)
                }));
                
                match bus.publish(message).await {
                    Ok(()) => {
                        published_count_clone.fetch_add(1, Ordering::Relaxed);
                        message_id += 1;
                    }
                    Err(e) => {
                        eprintln!("发布失败: {}", e);
                    }
                }
            }
        });
        
        published_count
    }

    /// 启动持续消费任务
    async fn start_continuous_consumer(
        bus: Arc<InMemoryMessageBus>,
        pattern: &str,
        stop_signal: Arc<AtomicBool>
    ) -> Arc<AtomicUsize> {
        let consumed_count = Arc::new(AtomicUsize::new(0));
        let consumed_count_clone = consumed_count.clone();
        let pattern = pattern.to_string();
        
        tokio::spawn(async move {
            let subscriber = match bus.subscribe(&pattern).await {
                Ok(sub) => sub,
                Err(e) => {
                    eprintln!("订阅失败: {}", e);
                    return;
                }
            };
            
            while !stop_signal.load(Ordering::Relaxed) {
                match subscriber.receive_timeout(1000).await {
                    Ok(Some(_)) => {
                        consumed_count_clone.fetch_add(1, Ordering::Relaxed);
                    }
                    Ok(None) => {
                        // 超时，继续
                    }
                    Err(_) => break,
                }
            }
            
            let _ = subscriber.close().await;
        });
        
        consumed_count
    }

    /// 监控系统状态
    async fn monitor_system_stats(
        bus: Arc<InMemoryMessageBus>,
        interval_seconds: u64,
        stop_signal: Arc<AtomicBool>
    ) -> Vec<SystemSnapshot> {
        let snapshots = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let snapshots_clone = snapshots.clone();
        let stop_signal_clone = stop_signal.clone();
        
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(interval_seconds));
            
            while !stop_signal_clone.load(Ordering::Relaxed) {
                interval.tick().await;
                
                let stats = match bus.stats().await {
                    Ok(s) => s,
                    Err(_) => continue,
                };
                
                let snapshot = SystemSnapshot {
                    timestamp: chrono::Utc::now(),
                    total_published: stats.total_published,
                    total_received: stats.total_received,
                    active_subscribers: stats.active_subscribers,
                    // 这里可以添加更多系统指标
                };
                
                snapshots_clone.lock().await.push(snapshot);
            }
        });
        
        // 等待停止信号，然后返回快照
        while !stop_signal.load(Ordering::Relaxed) {
            sleep(Duration::from_millis(100)).await;
        }
        
        Arc::try_unwrap(snapshots).unwrap().into_inner()
    }
}

/// 系统快照，用于监控长时间运行状态
#[derive(Debug, Clone)]
struct SystemSnapshot {
    timestamp: chrono::DateTime<chrono::Utc>,
    total_published: u64,
    total_received: u64,
    active_subscribers: usize,
}

impl SystemSnapshot {
    /// 计算吞吐量（消息/秒）
    fn throughput_since(&self, previous: &SystemSnapshot) -> f64 {
        let duration = self.timestamp.signed_duration_since(previous.timestamp);
        let duration_secs = duration.num_seconds() as f64;
        
        if duration_secs > 0.0 {
            (self.total_published - previous.total_published) as f64 / duration_secs
        } else {
            0.0
        }
    }
}

/// 长时间运行测试模块
mod long_running_tests {
    use super::*;

    #[tokio::test]
    async fn test_sustained_load_30_minutes() {
        println!("⏱️ 测试30分钟持续负载");
        
        let bus = LongRunningTestHelper::create_long_running_bus().await;
        let stop_signal = Arc::new(AtomicBool::new(false));
        
        // 测试参数
        let test_duration = Duration::from_secs(30 * 60); // 30分钟
        let messages_per_second = 100; // 每秒100条消息
        let expected_total_messages = messages_per_second * test_duration.as_secs() as usize;
        
        println!("📊 测试参数:");
        println!("  • 测试时长: {} 分钟", test_duration.as_secs() / 60);
        println!("  • 消息速率: {} msg/s", messages_per_second);
        println!("  • 预期总消息数: {}", expected_total_messages);
        
        // 启动发布者
        let publisher_count = LongRunningTestHelper::start_continuous_publisher(
            bus.clone(),
            "sustained.load",
            messages_per_second,
            stop_signal.clone()
        ).await;
        
        // 启动消费者
        let consumer_count = LongRunningTestHelper::start_continuous_consumer(
            bus.clone(),
            "sustained.load.*",
            stop_signal.clone()
        ).await;
        
        // 启动监控
        let monitor_stop = stop_signal.clone();
        let monitor_bus = bus.clone();
        let monitor_handle = tokio::spawn(async move {
            LongRunningTestHelper::monitor_system_stats(
                monitor_bus,
                60, // 每分钟记录一次
                monitor_stop
            ).await
        });
        
        // 等待测试时间
        println!("🚀 开始30分钟持续负载测试...");
        let start_time = Instant::now();
        
        // 定期报告进度
        let mut progress_interval = interval(Duration::from_secs(5 * 60)); // 每5分钟报告一次
        let mut last_published = 0;
        let mut last_consumed = 0;
        
        loop {
            tokio::select! {
                _ = progress_interval.tick() => {
                    let elapsed = start_time.elapsed();
                    let current_published = publisher_count.load(Ordering::Relaxed);
                    let current_consumed = consumer_count.load(Ordering::Relaxed);
                    
                    println!("📊 进度报告 ({:.1} 分钟):", elapsed.as_secs_f64() / 60.0);
                    println!("  • 已发布: {} (+{})", current_published, current_published - last_published);
                    println!("  • 已消费: {} (+{})", current_consumed, current_consumed - last_consumed);
                    
                    last_published = current_published;
                    last_consumed = current_consumed;
                }
                _ = sleep(test_duration) => {
                    break;
                }
            }
        }
        
        // 停止所有任务
        stop_signal.store(true, Ordering::Relaxed);
        let snapshots = monitor_handle.await.unwrap();
        
        // 最终统计
        let final_published = publisher_count.load(Ordering::Relaxed);
        let final_consumed = consumer_count.load(Ordering::Relaxed);
        
        println!("✅ 30分钟测试完成!");
        println!("📊 最终统计:");
        println!("  • 总发布消息: {}", final_published);
        println!("  • 总消费消息: {}", final_consumed);
        println!("  • 平均发布速率: {:.2} msg/s", final_published as f64 / test_duration.as_secs_f64());
        println!("  • 消息处理率: {:.2}%", (final_consumed as f64 / final_published as f64) * 100.0);
        
        // 分析快照
        if snapshots.len() >= 2 {
            let first = &snapshots[0];
            let last = &snapshots[snapshots.len() - 1];
            let avg_throughput = last.throughput_since(first);
            
            println!("  • 监控周期平均吞吐量: {:.2} msg/s", avg_throughput);
        }
        
        // 验证结果
        assert!(final_published >= expected_total_messages * 90 / 100, 
                "应该发布至少90%的预期消息数");
        assert!(final_consumed >= final_published * 80 / 100, 
                "应该消费至少80%的发布消息");
        
        // 验证系统状态
        let final_stats = bus.stats().await.unwrap();
        assert!(final_stats.total_published >= final_published as u64);
    }

    #[tokio::test]
    async fn test_memory_stability_over_time() {
        println!("🧠 测试长时间运行内存稳定性");
        
        let bus = LongRunningTestHelper::create_long_running_bus().await;
        let test_duration = Duration::from_secs(10 * 60); // 10分钟测试
        let stop_signal = Arc::new(AtomicBool::new(false));
        
        // 启动多个发布者和消费者
        let mut publisher_counts = Vec::new();
        let mut consumer_counts = Vec::new();
        
        // 创建5个发布者
        for i in 0..5 {
            let count = LongRunningTestHelper::start_continuous_publisher(
                bus.clone(),
                &format!("memory.test.{}", i),
                50, // 每个发布者每秒50条消息
                stop_signal.clone()
            ).await;
            publisher_counts.push(count);
        }
        
        // 创建5个消费者
        for i in 0..5 {
            let count = LongRunningTestHelper::start_continuous_consumer(
                bus.clone(),
                &format!("memory.test.{}.*", i),
                stop_signal.clone()
            ).await;
            consumer_counts.push(count);
        }
        
        // 记录初始内存状态
        let initial_stats = bus.stats().await.unwrap();
        
        println!("📊 内存稳定性测试开始:");
        println!("  • 测试时长: {} 分钟", test_duration.as_secs() / 60);
        println!("  • 发布者数量: {}", publisher_counts.len());
        println!("  • 消费者数量: {}", consumer_counts.len());
        println!("  • 初始活跃订阅者: {}", initial_stats.active_subscribers);
        
        // 定期检查内存使用情况
        let start_time = Instant::now();
        let mut check_interval = interval(Duration::from_secs(60)); // 每分钟检查一次
        let mut memory_samples = Vec::new();
        
        loop {
            tokio::select! {
                _ = check_interval.tick() => {
                    let elapsed = start_time.elapsed();
                    let stats = bus.stats().await.unwrap();
                    
                    let total_published: usize = publisher_counts.iter()
                        .map(|c| c.load(Ordering::Relaxed))
                        .sum();
                    let total_consumed: usize = consumer_counts.iter()
                        .map(|c| c.load(Ordering::Relaxed))
                        .sum();
                    
                    memory_samples.push((
                        elapsed,
                        stats.total_published,
                        stats.active_subscribers,
                        total_published,
                        total_consumed
                    ));
                    
                    println!("📊 内存检查 ({:.1} 分钟):", elapsed.as_secs_f64() / 60.0);
                    println!("  • 总发布: {}", total_published);
                    println!("  • 总消费: {}", total_consumed);
                    println!("  • 活跃订阅者: {}", stats.active_subscribers);
                }
                _ = sleep(test_duration) => {
                    break;
                }
            }
        }
        
        // 停止所有任务
        stop_signal.store(true, Ordering::Relaxed);
        
        // 等待一段时间让清理完成
        sleep(Duration::from_secs(5)).await;
        
        let final_stats = bus.stats().await.unwrap();
        
        println!("✅ 内存稳定性测试完成!");
        println!("📊 内存分析:");
        
        // 分析内存趋势
        if memory_samples.len() >= 3 {
            let first_sample = &memory_samples[0];
            let last_sample = &memory_samples[memory_samples.len() - 1];
            
            let throughput_change = (last_sample.1 - first_sample.1) as f64 
                / (last_sample.0.as_secs_f64() - first_sample.0.as_secs_f64());
            
            println!("  • 平均吞吐量: {:.2} msg/s", throughput_change);
            println!("  • 最终活跃订阅者: {}", final_stats.active_subscribers);
            
            // 检查内存是否稳定（订阅者数量不应该持续增长）
            assert_eq!(final_stats.active_subscribers, consumer_counts.len(),
                      "订阅者数量应该保持稳定");
        }
        
        // 验证系统仍然响应
        let test_message = Message::new("memory.test.final", json!({"test": "final"}));
        assert!(bus.publish(test_message).await.is_ok(), "系统应该仍然响应");
    }

    #[tokio::test]
    async fn test_subscriber_churn() {
        println!("🔄 测试订阅者频繁创建和销毁");
        
        let bus = LongRunningTestHelper::create_long_running_bus().await;
        let test_duration = Duration::from_secs(5 * 60); // 5分钟测试
        let start_time = Instant::now();
        
        let mut total_subscribers_created = 0;
        let mut total_subscribers_closed = 0;
        
        println!("📊 订阅者流失测试:");
        println!("  • 测试时长: {} 分钟", test_duration.as_secs() / 60);
        
        while start_time.elapsed() < test_duration {
            // 批量创建订阅者
            let mut subscribers = Vec::new();
            
            for i in 0..100 {
                let topic = format!("churn.test.{}", i % 10);
                match bus.subscribe(&topic).await {
                    Ok(subscriber) => {
                        subscribers.push(subscriber);
                        total_subscribers_created += 1;
                    }
                    Err(e) => {
                        eprintln!("创建订阅者失败: {}", e);
                    }
                }
            }
            
            // 使用订阅者一段时间
            sleep(Duration::from_millis(500)).await;
            
            // 发布一些测试消息
            for i in 0..10 {
                let topic = format!("churn.test.{}", i);
                let message = Message::new(&topic, json!({"test": i}));
                let _ = bus.publish(message).await;
            }
            
            // 让订阅者接收消息
            sleep(Duration::from_millis(500)).await;
            
            // 关闭所有订阅者
            for subscriber in subscribers {
                if subscriber.close().await.is_ok() {
                    total_subscribers_closed += 1;
                }
            }
            
            // 检查系统状态
            if start_time.elapsed().as_secs() % 60 == 0 {
                let stats = bus.stats().await.unwrap();
                println!("📊 流失测试进度 ({:.1} 分钟):", start_time.elapsed().as_secs_f64() / 60.0);
                println!("  • 已创建订阅者: {}", total_subscribers_created);
                println!("  • 已关闭订阅者: {}", total_subscribers_closed);
                println!("  • 当前活跃订阅者: {}", stats.active_subscribers);
            }
            
            // 短暂休息
            sleep(Duration::from_millis(100)).await;
        }
        
        // 最终检查
        sleep(Duration::from_secs(2)).await; // 等待清理完成
        let final_stats = bus.stats().await.unwrap();
        
        println!("✅ 订阅者流失测试完成!");
        println!("📊 最终统计:");
        println!("  • 总创建订阅者: {}", total_subscribers_created);
        println!("  • 总关闭订阅者: {}", total_subscribers_closed);
        println!("  • 最终活跃订阅者: {}", final_stats.active_subscribers);
        
        // 验证没有内存泄漏
        assert!(final_stats.active_subscribers <= 10, 
                "活跃订阅者数量应该很少，表明没有严重的内存泄漏");
        assert!(total_subscribers_created > 1000, 
                "应该创建了大量订阅者来测试系统稳定性");
    }

    #[tokio::test]
    async fn test_gradual_load_increase() {
        println!("📈 测试负载逐步增加");
        
        let bus = LongRunningTestHelper::create_long_running_bus().await;
        let test_phases = 5; // 5个阶段
        let phase_duration = Duration::from_secs(2 * 60); // 每个阶段2分钟
        
        println!("📊 负载递增测试:");
        println!("  • 测试阶段数: {}", test_phases);
        println!("  • 每阶段时长: {} 分钟", phase_duration.as_secs() / 60);
        
        let mut all_publishers = Vec::new();
        let mut all_consumers = Vec::new();
        
        for phase in 1..=test_phases {
            let messages_per_second = phase * 50; // 负载逐步增加
            
            println!("🚀 阶段 {}/{}：目标速率 {} msg/s", phase, test_phases, messages_per_second);
            
            let stop_signal = Arc::new(AtomicBool::new(false));
            
            // 启动这个阶段的发布者
            let publisher_count = LongRunningTestHelper::start_continuous_publisher(
                bus.clone(),
                &format!("load.increase.phase{}", phase),
                messages_per_second,
                stop_signal.clone()
            ).await;
            
            // 启动这个阶段的消费者
            let consumer_count = LongRunningTestHelper::start_continuous_consumer(
                bus.clone(),
                &format!("load.increase.phase{}.*", phase),
                stop_signal.clone()
            ).await;
            
            all_publishers.push((stop_signal.clone(), publisher_count));
            all_consumers.push((stop_signal, consumer_count));
            
            // 运行这个阶段
            let start_time = Instant::now();
            let mut report_interval = interval(Duration::from_secs(30));
            
            while start_time.elapsed() < phase_duration {
                report_interval.tick().await;
                
                let stats = bus.stats().await.unwrap();
                let total_published: usize = all_publishers.iter()
                    .map(|(_, count)| count.load(Ordering::Relaxed))
                    .sum();
                let total_consumed: usize = all_consumers.iter()
                    .map(|(_, count)| count.load(Ordering::Relaxed))
                    .sum();
                
                println!("  📊 阶段 {} 进度: 发布 {}, 消费 {}, 活跃订阅者 {}", 
                        phase, total_published, total_consumed, stats.active_subscribers);
            }
            
            println!("  ✅ 阶段 {} 完成", phase);
        }
        
        // 停止所有任务
        for (stop_signal, _) in &all_publishers {
            stop_signal.store(true, Ordering::Relaxed);
        }
        
        // 等待清理
        sleep(Duration::from_secs(3)).await;
        
        // 最终统计
        let final_stats = bus.stats().await.unwrap();
        let total_published: usize = all_publishers.iter()
            .map(|(_, count)| count.load(Ordering::Relaxed))
            .sum();
        let total_consumed: usize = all_consumers.iter()
            .map(|(_, count)| count.load(Ordering::Relaxed))
            .sum();
        
        println!("✅ 负载递增测试完成!");
        println!("📊 总体统计:");
        println!("  • 总发布消息: {}", total_published);
        println!("  • 总消费消息: {}", total_consumed);
        println!("  • 消息处理率: {:.2}%", (total_consumed as f64 / total_published as f64) * 100.0);
        println!("  • 最终系统状态: {} 活跃订阅者", final_stats.active_subscribers);
        
        // 验证系统在高负载下仍然稳定
        assert!(total_published > 10000, "应该发布了大量消息");
        assert!(total_consumed >= total_published * 70 / 100, "消息处理率应该 >= 70%");
        
        // 验证系统仍然响应
        let test_message = Message::new("load.test.final", json!({"test": "final"}));
        assert!(bus.publish(test_message).await.is_ok(), "高负载后系统应该仍然响应");
    }
}